package com.shujia.flink.source;

import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class Demo01SocketSource {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(4);

        // Socket Source只能由1个并行度去接收数据
        DataStreamSource<String> socketDS = env.socketTextStream("master", 8888);// 无界流
        System.out.println(socketDS.getParallelism());

        // 上游的数据是轮询过来的
        //        socketDS.print();


        // 如果对流进行了keyBy操作，则数据不再是轮询发送到下游，而是以Hash取余的原理计算到底被哪个线程处理
        socketDS.keyBy(w -> w).print();


        env.execute();

    }
}
